package chan.module.project.hw.mrs.v350;

import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Scanner;

public class MRSHdfsV350Main {
    private final static Log LOG = LogFactory.getLog(MRSHdfsV350Main.class.getName());
    private static final String PATH_TO_HDFS_SITE_XML = "/data/tongtech/TongDXP3181/Node-Agent/mrs-all-config/HDFS/config/hdfs-site.xml";
    private static final String PATH_TO_CORE_SITE_XML = "/data/tongtech/TongDXP3181/Node-Agent/mrs-all-config/Hive/config/core-site.xml";
    private static final String PATH_TO_KEY_TAB = "/data/tongtech/TongDXP3181/Node-Agent/mrs-all-config/swap_mrs_uat.keytab";
    private static final String PATH_TO_KRB5_CONF = "/data/tongtech/TongDXP3181/Node-Agent/mrs-all-config/KrbClient/config/krb5.conf";
    private static final String PRINCIPAL_NAME = "swap_mrs_uat@HADOOP.COM";
    private static FileSystem fSystem;

    public static void main(String[] args) throws Exception {
        System.setProperty("java.security.krb5.conf", PATH_TO_KRB5_CONF);
        Configuration conf = new Configuration();
        conf.addResource(new Path(PATH_TO_HDFS_SITE_XML));
        conf.addResource(new Path(PATH_TO_CORE_SITE_XML));
        LoginUtil.login(PRINCIPAL_NAME, PATH_TO_KEY_TAB, PATH_TO_KRB5_CONF, conf);
        fSystem = FileSystem.get(conf);
        Scanner sc = new Scanner(System.in);
        String in;
        while (true){
            LOG.info("1-创建HDFS目录；2-从本地上传文件到HDFS；3-读取HDFS的文本文件；其他退出");
            in = sc.next();
            try {
                switch (in){
                    case "1" : {
                        LOG.info("请输入需要创建的HDFS目录：");
                        in = sc.next();
                        LOG.info("准备创建的HDFS目录：" + in);
                        createPath(in);
                        break;
                    }
                    case "2" : {
                        LOG.info("请输入上传的本地文件路径：");
                        String path_from_local = sc.next();
                        LOG.info("请输入HDFS文件路径：");
                        String path_to_hdfs = sc.next();
                        LOG.info(" 本地文件路径:"+path_from_local+" , HDFS文件路径: "+path_to_hdfs);
                        write(path_from_local,path_to_hdfs);
                        break;
                    }
                    case "3" : {
                        LOG.info("请输入需要读取的HDFS文件路径：");
                        String path_to_hdfs = sc.next();
                        read(path_to_hdfs);
                        break;
                    }
                    default:
                        LOG.info("退出程序！");
                        return;
                }
            }catch (Exception e){
                LOG.error("执行失败！",e);
            }

        }
    }

    /**
     * 创建文件
     * @param path_to_hdfs
     * @return
     * @throws IOException
     */
    private static  boolean createPath(String path_to_hdfs) throws IOException {
        Path filePath = new Path(path_to_hdfs);
        if (!fSystem.exists(filePath)) {
            fSystem.mkdirs(filePath);
        }
        return true;
    }


    private static void write(String path_from_local,String path_to_hdfs ) throws IOException {
        File file = new File(path_from_local);
        byte[] fileContent = FileUtils.readFileToByteArray(file);
        FSDataOutputStream out = null;
        try {
            out = fSystem.create(new Path(path_to_hdfs));
            out.write(fileContent);
            out.hsync();
            LOG.info("success to write.");
        } finally {
            // make sure the stream is closed finally.
            IOUtils.closeStream(out);
        }
    }

    private static void read(String path_to_hdfs) throws IOException {
        Path path = new Path(path_to_hdfs);
        FSDataInputStream in = null;
        BufferedReader reader = null;
        StringBuffer strBuffer = new StringBuffer();
        try {
            in = fSystem.open(path);
            reader = new BufferedReader(new InputStreamReader(in));
            String sTempOneLine;
            // write file
            while ((sTempOneLine = reader.readLine()) != null) {
                strBuffer.append(sTempOneLine);
            }
            LOG.info("result is : " + strBuffer.toString());
            LOG.info("success to read.");
        } finally {
            // make sure the streams are closed finally.
            IOUtils.closeStream(reader);
            IOUtils.closeStream(in);
        }
    }

}